/*
 *  DataRepository.cpp
 *  MetaServer
 *
 *  Created by Ian Pye on 1/7/08.
 *  Copyright 2008 TriceratopsPowered. All rights reserved.
 *
 */

// STD lib
#include <iostream>
#include <string> 
#include <typeinfo>

#include <fcntl.h>
#include <sys/stat.h>
#include <sys/types.h>

// Loggers
#include <log4cxx/logger.h>
#include <log4cxx/basicconfigurator.h>
#include <log4cxx/helpers/exception.h>
#include <log4cxx/propertyconfigurator.h>

// Thrift
#include <transport/TServerSocket.h>
#include <protocol/TBinaryProtocol.h>
#include <transport/TTransportUtils.h>

// berkeleydb
#include <db_cxx.h>

// Mah Own
#include "Ringer.h"
#include "RingerMetaServer.h"
#include "DataRepository.h"

using namespace std;
using namespace ringer;

using namespace facebook::thrift;
using namespace facebook::thrift::protocol;
using namespace facebook::thrift::transport;


using boost::shared_ptr;

static int sem_oflags = O_CREAT;
static mode_t sem_mode = 0777;

static u_int32_t g_mds_db_oFlags = DB_CREATE; // Open flags;
static u_int32_t g_mds_db_env_oFlags = DB_CREATE | DB_INIT_MPOOL; // Initialize the in-memory cache.
static DbEnv::DbEnv g_ringer_mds_env(0);        // DB Enviroment   

static const int MAX_SERIALIZED_SIZE = 32768;

/**
 Setup a simple file based semaphore
 */
DataRepository::DataRepository(std::string db_env, std::string db_file_name){ 
  
  // Init semmaphore
  pm_permissions_sem = sem_open(PERMISSIONS_SEM_NAME.c_str(), sem_oflags, sem_mode, 0);
  
  // And, setup the protocol factory!
  mp_protocolFactory = new boost::shared_ptr<TProtocolFactory>(new TBinaryProtocolFactory());
  
  // Finally, open the db
  try {
    g_ringer_mds_env.open(db_env.c_str(),  g_mds_db_env_oFlags, 0);
    mds_db = new Db(&g_ringer_mds_env, 0);
    mds_db->open(NULL,
                 db_file_name.c_str(),
                 NULL,
                 DB_BTREE,
                 g_mds_db_oFlags,
                 0);
    LOG4CXX_INFO(g_metaserver_logger, "DB online.");             
  } catch(DbException &e) {
    LOG4CXX_FATAL(g_metaserver_logger, (string)("Error opening db: ")+e.what());
    exit( -1 );
  } catch(std::exception &e) {
    LOG4CXX_FATAL(g_metaserver_logger, (string)("Error opening db: ")+e.what());
    exit( -1 );
  }
}

DataRepository::~DataRepository(){  
  sem_close( pm_permissions_sem );
  sleep(1); // let the other processes get out of the way
  sem_unlink( PERMISSIONS_SEM_NAME.c_str() );
  
  // close the db
  try {
    if (mds_db != NULL) {
      mds_db->close(0);
    }
    g_ringer_mds_env.close(0);
  } catch(DbException &e) {
    LOG4CXX_FATAL(g_metaserver_logger, (string)("Error closing db: ")+e.what());
    exit( -1 );
  } catch(std::exception &e) {
    LOG4CXX_FATAL(g_metaserver_logger, (string)("Error closing db: ")+e.what());
    exit( -1 );
  }
}

/*
 Get the rnode from the db
 */
int DataRepository::get(const rpath_t& path, RNode& rn){
  shared_ptr<TMemoryBuffer> mbuf(new TMemoryBuffer());
  Dbt data;
  
  // Get it from persistant storage
  Dbt key(const_cast<char *>(path.c_str()), path.length()+1);
  data.set_data(mbuf->getWritePtr(MAX_SERIALIZED_SIZE));
  data.set_ulen(MAX_SERIALIZED_SIZE);
  data.set_flags(DB_DBT_USERMEM);
  
  try{
    if (mds_db->get(NULL, &key, &data, 0) == DB_NOTFOUND){
      LOG4CXX_INFO(g_metaserver_logger, "Not found: "+path);
      return RNOODE_NOT_FOUND;  // return empty
    }
  }
  catch (DbException dbex){
    LOG4CXX_WARN(g_metaserver_logger, "Error Getting " + path + ": " +dbex.what());
  }
  mbuf->wroteBytes(MAX_SERIALIZED_SIZE);
  
  rn.read(get_pointer((*mp_protocolFactory)->getProtocol(mbuf)));
  
  return 0;    
}

/**
 Write the rnode to the database
 */
bool DataRepository::put(const RNode& rn){
  
  shared_ptr<TMemoryBuffer> mbuf(new TMemoryBuffer());
  rn.write(get_pointer((*mp_protocolFactory)->getProtocol(mbuf)));
  
  // bdb key, data entries 
  Dbt key(const_cast<char *>(rn.path.c_str()),  rn.path.length()+1);
  std::string serialized = mbuf->getBufferAsString();
  Dbt data(const_cast<char *>(serialized.c_str()), serialized.length()+1);
  
  // stick it in the db
  int ret = mds_db->put(NULL, &key, &data, 0);
  if (ret != 0){
    LOG4CXX_WARN(g_metaserver_logger, ("Error inserting to db: "));
    return false;
  }
  
  return true;
}

/**
 Remove the rnode from the DB.
 */
bool DataRepository::r_delete(const rpath_t& path){  
  Dbt key(const_cast<char *>(path.c_str()), path.length()+1);
  if (mds_db->del(NULL, &key, 0) == DB_NOTFOUND){
    LOG4CXX_INFO(g_metaserver_logger, "Error deleteing " + path + ": not found");
    return false;
  }
  LOG4CXX_INFO(g_metaserver_logger, "Deleted "+path);
  return true;
}

/**
 Gets a read priveledge
 */
int DataRepository::get_read(const rpath_t& path, const UserInfo& userInfo){
  RNode rn;
  int res;
  res = sem_trywait( pm_permissions_sem );
  
  if (get(path, rn) < 0)
    return RNOODE_NOT_FOUND;
  
  rn.readers.insert(userInfo.ringer_uid);
  put(rn);
  res = sem_post(pm_permissions_sem);
  return 0;
}
int DataRepository::get_write(const rpath_t& path, const UserInfo& userInfo){
  RNode rn;
  int res;
  set<int32_t>::iterator first;
  set<int32_t>::iterator last;
  
  res = sem_trywait( pm_permissions_sem );
  
  if (get(path, rn) < 0)
    return RNOODE_NOT_FOUND;
  
  if (rn.writer != NO_USER && rn.writer != userInfo.ringer_uid){
    string logstr = "Denyied write: writer = ";
    logstr += rn.writer;
    logstr += ". ringer_uid = ";
    logstr += userInfo.ringer_uid;
    LOG4CXX_INFO(g_metaserver_logger, logstr);
    return RNOODE_NOT_ALLOWED;
  }
  
  rn.writer = userInfo.ringer_uid;
  first = rn.readers.begin();
  last = rn.readers.end();
  rn.readers.erase(first, last);  
  put(rn);
  
  res = sem_post(pm_permissions_sem);
  
  return 0;
}
bool DataRepository::end_read(const rpath_t& path, const UserInfo& userInfo){
  RNode rn;
  if (!get(path, rn))
    return false;
  
  if (rn.readers.erase(userInfo.ringer_uid) < 1)
    return false;
  
  put(rn);
  return true;
}
bool DataRepository::end_write(const rpath_t& path, const UserInfo& userInfo){
  RNode rn;
  if (!get(path, rn))
    return false;
  
  if (rn.writer != userInfo.ringer_uid)
    return false;
  
  rn.writer = NO_USER;  
  put(rn);
  return true;
}
